package com.example.mq_sm.mqserver.datacenter;

import com.example.mq_sm.common.MQException;
import com.example.mq_sm.mqserver.core.Binding;
import com.example.mq_sm.mqserver.core.Exchange;
import com.example.mq_sm.mqserver.core.MSGQueue;
import com.example.mq_sm.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 使用这个类来统一管理内存中的所有数据
 * 该类后续提供一些方法，可能会在多线程环境下使用，因此要注意线程安全问题
 */
public class MemoryDataCenter {
    // key 是 exchangeName, value 是 Exchange 对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // key 是 queueName, value 是 MSGQueue 对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 第一个 key 是 exchangeName, 第二个 key 是 queueName
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // key 是 messageId, value 是 Message 对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // key 是 queueName, value 是一个 Message 的链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 第一个 key 是 queueName， 第二个 key 是 messageId
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(), exchange);
    }

    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
    }

    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(), queue);
    }

    public MSGQueue getQueue(String queueName){
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
    }

    public void insertBinding(Binding binding) throws MQException {
        // 先使用 exchangeName 查一下，对应的哈细胞是否存在，不存在就创建一个
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());
        // 这段代码等效为
//        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
//        if (bindingMap == null){
//            bindingMap = new ConcurrentHashMap<>();
//            bindingMap.put(binding.getExchangeName(), bindingMap);
//        }
        synchronized (bindingMap){
            // 再进行一次检查
            if (bindingMap.get(binding.getQueueName()) != null){
                throw new MQException("[MemoryDataCenter] 绑定已存在！exchangeName=" + binding.getExchangeName() +
                        ", queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 新绑定添加成功！exchangeName=" + binding.getExchangeName() +
                ", queueName=" + binding.getQueueName());
    }

    // 获取绑定, 写两个版本
    // 1. 根据 exchangeName 和 queueName 确定唯一一个 Binding
    // 2. 根据 exchangeName 获取到所有的 Binding
    public Binding getBinding(String exchangeName, String queueName){
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null){
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MQException {
        String queueName = binding.getQueueName();
        String exchangeName = binding.getExchangeName();
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null){
            throw new MQException("[MemoryDataCenter] 绑定已存在！exchangeName=" + binding.getExchangeName() +
                    ", queueName=" + binding.getQueueName());
        }
        bindingMap.remove(queueName);
        System.out.println();
    }

    // 添加消息
    public void addMessage(Message message){
        messageMap.put(message.getMessageId(), message);
    }

    // 根据 id 删除消息
    public void removeMessage(String messageId){
        messageMap.remove(messageId);
    }

    // 发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message){
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(),
                k -> new LinkedList<>());
        synchronized (messages){
            messages.add(message);
        }
        addMessage(message);
        System.out.println("[MemoryDataCenter] 消息被投递到队列中！messageId=" + message.getMessageId());
    }

    // 从队列中取消息
    public Message pollMessage(String queueName){
        // 根据队列名查找一下对应的队列的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null){
            return null;
        }
        synchronized (messages){
            if (messages.size() == 0){
                return null;
            }
            // 链表有元素，就进行头删
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出！messageId=" + currentMessage.getMessageId());
            return currentMessage;
        }
    }

    // 获取指定队列中消息的个数
    public int getMessageCount(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null){
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }

    // 添加未确认的消息
    public void addMessageWaitAck(String queueName, Message message){
        ConcurrentHashMap<String, Message> messageConcurrentHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageConcurrentHashMap.put(message.getMessageId(), message);
    }

    // 删除未确认的消息（消息已经确认）
    public void removeMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messageConcurrentHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageConcurrentHashMap == null){
            return;
        }
        messageConcurrentHashMap.remove(messageId);
    }

    // 获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messageConcurrentHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageConcurrentHashMap == null){
            return null;
        }
        return messageConcurrentHashMap.get(messageId);
    }

    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }

    // 从硬盘上读取数据，把硬盘中之前持久化存储的各个维度的数据数据都恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws MQException, IOException, ClassNotFoundException {
        // 先对所有数据进行一个清空
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        //DiskDataCenter diskDataCenter = new DiskDataCenter();
        // 恢复 exchange 的数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges){
            exchangeMap.put(exchange.getName(), exchange);
        }

        // 恢复 queue 和 message 的数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue : queues){
            queueMap.put(queue.getName(), queue);
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            for (Message message : messages){
                messageMap.put(message.getMessageId(), message);
            }
        }

        // 恢复 binding 的数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for(Binding binding : bindings){
            String exchangeName = binding.getExchangeName();
            ConcurrentHashMap<String, Binding> tmp = bindingsMap.computeIfAbsent(exchangeName,
                    k -> new ConcurrentHashMap<>());
            //bindingsMap.put(exchangeName, tmp);
            tmp.put(binding.getQueueName(), binding);
        }

        // 恢复 message 的数据
        //LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue()
    }
}


















